package com.zyh;



import com.zyh.entity.MyIntWritable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

public class WordCountJob extends Configured implements Tool {

    public static void main(String[] args) throws Exception{
        ToolRunner.run(new WordCountJob(),args);
    }

    @Override
    public int run(String[] strings) throws Exception {
        /**
         * 组装job任务
         */
        //String input = "/mr/demo1/user-pay-2021-9-9.log";
        String input = "/mr/demo1/product.txt";
        String output = "/mr/out/demo3/";

        //1 初始化配置
        Configuration conf = new Configuration();//自动加载classpath下的配置文件。
        conf.set("fs.defaultFS","hdfs://192.168.193.10:9000");

        //2 创建job
        Job job = Job.getInstance(conf);
        job.setJarByClass(WordCountJob.class);

        //3 设置输入
        FileInputFormat.setInputPaths(job,new Path(input));

        //4 设置mapper
        job.setMapperClass(WordCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(MyIntWritable.class);
        //job.setCombinerClass(WordCountReducer.class);//在map阶段提前执行reduce方法
        //5 设置reducer
        job.setReducerClass(WordCountReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        //job.setNumReduceTasks(2);//将reduce数量设置为2

        //6 设置输出路径(输出路径不能存在)
        FileOutputFormat.setOutputPath(job,new Path(output));
        //如果输出路径存在则删除。
        if(FileSystem.get(conf).exists(new Path(output))){
            FileSystem.get(conf).delete(new Path(output),true);
        }


        //7 启动job
        boolean b = job.waitForCompletion(true);
        return b == true ? 0:1;// 成功返回0 ，失败返回1
    }
}

/**
 * LongWritable：偏移量
 * Text：传递整行参数
 * Text：输出后的key为Text类型，字符串类型
 * IntWritable：输出结果value的类型
 */
class WordCountMapper extends Mapper<LongWritable,Text, Text, MyIntWritable> {
    private SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//        //把整行数据转换为key为name,值为1
//        String[] split = value.toString().split("\t");
//        String name = split[1];
//        //map运行完了要生成key和value，key是姓名，value是1
//        context.write(new Text(name),new IntWritable(1));
        /**
         * 统计11-10这天，每个商品的访问量。part-r-0000

        String[] split = value.toString().split("  ");
        String name = split[2];
        context.write(new Text(name),new IntWritable(1));
*/
        /**
         * 统计每个小时内，平台商品访问次数。

        String[] split = value.toString().split("  ");
        try {
            Date parse = simpleDateFormat.parse(split[0]);
            String hours = parse.getHours()+"";
            context.write(new Text(hours),new IntWritable(1));
        } catch (ParseException e) {
            e.printStackTrace();
        }*/

        /**
         * 统计每个商品的平均访问时间

        String[] s = value.toString().split("  ");
        Integer i = Integer.parseInt(s[4]);
        context.write(new Text(s[2]),new MyIntWritable(i));*/

        /**
         * 按照访问量做降序
         */
        String[] s = value.toString().split("  ");
        context.write(new Text(s[2]),new MyIntWritable(1));
    }
}
class WordCountReducer extends Reducer<Text,MyIntWritable,Text,IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<MyIntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;

        /**
         * 统计商品访问次数

        for (MyIntWritable value : values) {
            sum += value.getValue();
        }
        context.write(key,new IntWritable(sum));*/

        /**
         * 统计商品平均停留时间

        int length = 0;
        for (MyIntWritable value : values) {
            sum += value.getValue();
            length++;
        }
        context.write(key,new IntWritable(sum/length));*/

        /**
         * 按照访问量做降序
         */
        for (MyIntWritable value : values) {
            sum += value.getValue();
        }

    }
}